package nsq

import (
	"fmt"
	gconfig "funlog-gserver/config"
	nsqModel "funlog-gserver/model/nsq"
	"funlog-gserver/utils"
	"log"
	"sync"

	"github.com/nsqio/go-nsq"
)

//producer 消息生产者
var producer *nsq.Producer
var once sync.Once

//InitNSQ 单初始化消息队列
func InitNSQ() {
	once.Do(func() {
		//初始化生产者
		producer = initProducer()
	})
}

//initProducer 初始化生产者
func initProducer() *nsq.Producer {
	config := nsq.NewConfig()

	nsqAddr := gconfig.Cnf["NSQ_ADDR"]
	pd, err := nsq.NewProducer(nsqAddr, config)
	if err != nil {
		log.Printf("create producer failed, err:")
		log.Fatal(err)
	}
	return pd
}

//NewConsumers 定义一个消息队列消费者
func NewConsumers(topic string, channel string, handler nsq.Handler) error {
	addr := gconfig.Cnf["NSQ_LOOK_ADDR"]
	conf := nsq.NewConfig()
	nc, err := nsq.NewConsumer(topic, channel, conf)
	if err != nil {
		fmt.Println("create consumer failed err ", err)
		return err
	}
	nc.AddHandler(handler)
	// 连接nsqlookupd
	if err := nc.ConnectToNSQLookupd(addr); err != nil {
		fmt.Println("connect nsqlookupd failed ", err)
		return err
	}
	return nil
}

//PublishMsg 推送一个消息到消息队列
func PublishMsg(nsqMsg *nsqModel.NMessage) error {
	if producer != nil {
		message := utils.Strval(nsqMsg.Body)
		// nsqMsg
		return producer.Publish(nsqMsg.Topic, []byte(message))
	}
	return nil
}
